package com.lkh.map;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.types.Row;

import java.text.SimpleDateFormat;
import java.util.Date;

public class SentMapFunaction implements MapFunction<Row, Tuple4<String, String, Double, Integer>> {
    @Override
    public Tuple4<String, String, Double, Integer> map(Row value) throws Exception {

        String sentId = (String) value.getField(0);
        String time = (String) value.getField(1);
        Double prediction = (Double) value.getField(2);
        String predictionDetail = (String) value.getField(3);

        Date date = new Date(time);

        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd:hh");
        String day = format.format(date);

        JSONObject jsonObject = JSON.parseObject(predictionDetail);

        //正面赋值 0
        Double just = jsonObject.getDouble("0.0");
        //负面赋值 1
        Double lose = jsonObject.getDouble("1.0");

        //如果预测结果概率相差不多，将预测结果改成中性-2
        if (Math.abs(just - lose) < 0.3) {
            prediction = 2.0;
        }

        return new Tuple4<>(sentId, day, prediction, 1);
    }
}
